package com.lkx.kafka.monitor.controller;

import com.lkx.kafka.monitor.dto.ClusterMetrics;
import com.lkx.kafka.monitor.dto.ConsumerGroupMetrics;
import com.lkx.kafka.monitor.dto.TopicMetrics;
import com.lkx.kafka.monitor.service.KafkaAdminService;
import com.lkx.kafka.monitor.service.KafkaMonitorService;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.util.List;
import java.util.Map;

@RestController
@RequestMapping("/kafka/admin")
@RequiredArgsConstructor
@CrossOrigin(origins = "*") // 允许跨域访问，方便前端调试
public class KafkaAdminController {

    @Autowired
    private KafkaAdminService  kafkaAdminService;

    /**
     * 创建Kafka主题
     */
    @PostMapping("/topics")
    public ResponseEntity<Map<String, String>> createTopic(@RequestBody Map<String, Object> topicConfig) {
        try {
            String topicName = (String) topicConfig.get("name");
            int partitions = (Integer) topicConfig.get("partitions");
            int replicas = (Integer) topicConfig.get("replicas");
            int retentionMs = (Integer) topicConfig.get("retentionMs");

            kafkaAdminService.createTopic(topicName, partitions, replicas, retentionMs*1000L);

            return ResponseEntity.ok(Map.of(
                    "status", "success",
                    "message", "Topic " + topicName + " created successfully"
            ));
        } catch (Exception e) {
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                    .body(Map.of(
                            "status", "error",
                            "message", "Failed to create topic: " + e.getMessage()
                    ));
        }
    }

    /**
     * 删除Kafka主题
     */
    @DeleteMapping("/topics/{topicName}")
    public ResponseEntity<Map<String, String>> deleteTopic(@PathVariable String topicName) {
        try {
            kafkaAdminService.deleteTopic(topicName);

            return ResponseEntity.ok(Map.of(
                    "status", "success",
                    "message", "Topic " + topicName + " deleted successfully"
            ));
        } catch (Exception e) {
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                    .body(Map.of(
                            "status", "error",
                            "message", "Failed to delete topic: " + e.getMessage()
                    ));
        }
    }

    /**
     * 保存Kafka配置
     */
    @PostMapping("/config")
    public ResponseEntity<Map<String, String>> saveKafkaConfig(@RequestBody Map<String, Object> config) {
        try {
            // 这里应该调用相应的服务方法来保存配置
            // 目前只是示例实现
            return ResponseEntity.ok(Map.of(
                    "status", "success",
                    "message", "Kafka config saved successfully"
            ));
        } catch (Exception e) {
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                    .body(Map.of(
                            "status", "error",
                            "message", "Failed to save Kafka config: " + e.getMessage()
                    ));
        }
    }

}